/*******************************************************
  > Copyright (C) 2024 ==iot== All rights reserved.
  > File Name: 2创建线程池.c
  > Author: wyf
  > Created Time: 2024年08月09日 星期五 17时01分28秒
 ***************************************************/


#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <dirent.h> //目录名获取配合DIR
#include <assert.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>
#include <fcntl.h> //文件控制操作
#include <sys/types.h> //获取pid等
#include <sys/wait.h> //等待子进程结束
#include <time.h> //时间获取
#include <sys/ipc.h> //进程通信
#include <sys/msg.h> //消息队列
#include <sys/shm.h> //共享内存
#include <signal.h> //信号处理
#include <pthread.h> //线程
#include <semaphore.h>
struct exam
{
	int data1;
	int data2;
};//队列结构体的数据域

typedef struct queue
{
	struct exam content;
	struct queue* next;
}Queue;//队列结构体



struct threadpool
{
	sem_t semw;
	sem_t semr;
	pthread_mutex_t mutex;

	struct queue* que;
	pthread_t* tids;
	int queue_length_max;
};//线程池结构体

int flag = 0;

// 初始化头结点
Queue* create_queue() {
	Queue* head = (Queue*)malloc(sizeof(Queue));
	head->next = NULL;
	return head;
}
// 入队
void push(Queue* head, struct exam* data) {
	Queue* newNode = (Queue*)malloc(sizeof(Queue));
	newNode->content.data1 = data->data1;
	newNode->content.data2 = data->data2;
	Queue* p = head;
	while (p->next != NULL) {
		p = p->next;
	}

	p->next = newNode;
	newNode->next = NULL;
	p = newNode;
}
// 获取队列头的第一个数据
int front(Queue* head) {
	if (head->next == NULL) {
		printf("Queue is empty\n");
		return -1;
	}
	printf("当前队列中第一个数据是：%d\t", head->next->content.data1);
	printf("%d\n", head->next->content.data2);
	return 0;
}
// 出队
void pop(Queue* head) {
	if (head->next == NULL) {
		printf("队列是空的!\n");
		return;
	}
	Queue* p = head->next;
	head->next = head->next->next;
	free(p);
}
// 返回队列的大小
int size(Queue* head) {
	int count = 0;
	Queue* p = head->next;
	while (p != NULL) {
		count++;
		p = p->next;
	}
	return count;
}

// 判断队列是否为空
int isempty(Queue* head) {
	return head->next == NULL;  // 如果头结点的指针为空，返回1表示空；否则返回0
}

// 销毁队列
void destroy_queue(Queue* head) {
	Queue* p = head;
	while (p != NULL) {
		Queue* temp = p;
		p = p->next;
		free(temp);
	}
}



// 创建线程池
struct threadpool* create_threadpool(int threads, int queue_length, void* (*consumer)(void*))
{
	struct threadpool* p = malloc(sizeof(struct threadpool));//!!!

	p->tids = malloc(sizeof(pthread_t) * threads);//线程池的线程id
	p->que = create_queue();//队列的头结点
	int ret = 0;
	sem_init(&p->semw, 0, queue_length);
	sem_init(&p->semr, 0, 0);
	pthread_mutex_init(&p->mutex, NULL);

	for (int i = 0; i < threads; i++)//创建消费者线程
	{
		ret = pthread_create(&p->tids[i], NULL, consumer, p);
		if (ret != 0)
		{
			perror("pthread_create");
		}
	}

	return p;																					//这里是关键!!!
}

//销毁线程池
void destroy_threadpool(struct threadpool* p, int threads)
{
	for (int i = 0; i < threads; i++)
	{
		pthread_join(p->tids[i], NULL);
	}

	sem_destroy(&p->semw);
	sem_destroy(&p->semr);
	pthread_mutex_destroy(&p->mutex);
	free(p->tids);																			//释放线程id,malloc获取的！！！
	free(p);																				//释放结构体，也就是释放线程池！！！

	destroy_queue(p->que);//释放队列
	printf("已释放队列及销毁线程池\n");
}


//生产者线程
void* producer(void* arg)
{
	struct threadpool* p = (struct threadpool*)arg;//强转
	while (flag == 0)
	{

		for (int i = 0; i < p->queue_length_max; i++)									//生产者一次写入所有数据！！！
		{
			sem_wait(&p->semw);//阻塞
			if (flag == 1)
			{
				break;
			}
			pthread_mutex_lock(&p->mutex);//互斥锁
			printf("给队列里输入两个值:\n");
			scanf("%d", &p->que->content.data1);	//可以用针这样持续指向数据，也可以用一个新的数据变量k.data1再push给入队列q->que中！！！
			scanf("%d", &p->que->content.data2);

			push(p->que, &p->que->content);												//刚开始检测的第一步队列的头是正确的！！！
			printf("入队数据是：%d %d\n", p->que->content.data1, p->que->content.data2);

			pthread_mutex_unlock(&p->mutex);//释放锁
		}

		for (int i = 0; i < p->queue_length_max; i++)									//一次释放所有消费者线程！！！				
		{
			sem_post(&p->semr);//唤醒消费者线程											//必须是这个解锁顺序！！！
		}


	}
	printf("生产者线程退出\n");
}

//消费者线程
void* consumer(void* arg)
{
	int count = 0;
	struct threadpool* p = (struct threadpool*)arg;

	while (flag == 0)
	{

		sem_wait(&p->semr);//阻塞
		if (flag == 1)
		{
			break;
		}
		pthread_mutex_lock(&p->mutex);//互斥锁


		printf("\n消费者线程---%ld号在处理数据：\n", pthread_self() % 100);//获取线程id
		count = size(p->que);//发送队列头结点求大小
		printf("队列长度是：%d\n", count);
		front(p->que);//查找队列头结点的第一个数据
		pop(p->que);//出队


		pthread_mutex_unlock(&p->mutex);//释放锁
						//一对多以后，不用再告诉生产者进程了！！！

		sleep(1);

	}

	printf("消费者线程---%ld号退出\n",pthread_self()%100);
}

void fun(int sig)
{
	flag = 1;
}

int main(int argc, const char* argv[])
{
	signal(SIGINT, fun);

	int ret = 0;
	int threads = 0;
	int queue_length = 0;
	threads = 3;//控制消费者线程数量
	queue_length = 5;//队列长度控制的是一个生产者生产的数量						

	printf("线程池中消费者线程的数量:%d\n",threads);
	printf("队列长度:%d\n",queue_length);

	// 创建线程池---多消费者线程
	struct threadpool *p = create_threadpool(threads, queue_length, consumer);	//必选先激活多消费者线程!!!

	// 创建生产者线程
	pthread_t tid1 = 0;
	p->queue_length_max = queue_length;
	sem_init(&p->semw, 0, queue_length);//初始化信号量							指针下的信号量不提醒！！！
	sem_init(&p->semr, 0, 0);//初始化信号量										信号量在生产者消费者下面必须都初始化！！！
	pthread_mutex_init(&p->mutex, NULL);//初始化互斥锁
	p->que = create_queue();
	ret = pthread_create(&tid1, NULL, producer, p);//激活生产者线程
	if (ret != 0) {
		perror("pthread_create");
	}
	//一直在这等待信号SIGINT，然后唤醒让生产者和消费者线程都去可以检查flag！！！
	while (flag == 0)
	{
		sleep(1);
	}
	sem_post(&p->semw);//唤醒生产者线程
	//应该唤醒所有消费者线程退出!!!!
	for (int i = 0; i < threads; i++)
	{
		sem_post(&p->semr);//唤醒消费者线程
	}


	pthread_join(tid1, NULL);//等待生产者线程结束
	destroy_threadpool(p, threads);//销毁线程池

	return 0;

}


